"""
Case Type   : Logical_Replication
Case Name   : 逻辑复制数据精度传入和传出
Create At   : 2024/7/23
Owner       : @lvlintao666
Description :
    1.修改参数
    2.创建表
    3.插入数据
    4.修改表内索引
    5.修改表内数据
    6.查看数据
    7.清理表
    8.恢复参数，恢复pg_hba.conf文件
Expect      :
    1.修改参数成功
    2.创建成功
    3.插入成功
    4.修改索引成功
    5.修改数据成功
    6.查看数据成功
    7.清理成功
    7.恢复成功
History     :
"""
import os
import unittest
from yat.test import macro
from testcase.utils.Logger import Logger
from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant

class LogicReplication(unittest.TestCase):
    def setUp(self):
        self.logger = Logger()
        self.logger.info(f"-----{os.path.basename(__file__)} start-----")
        self.sh_primary = CommonSH('PrimaryDbUser')
        self.constant = Constant()
        self.common = Common()
        self.user_name = 'u_logical_replication_0087'
        self.slot_name = 'slot_logical_replication_0087'
        self.table_name = 't_logical_0087'
        self.config_wal_level = 'wal_level'
        self.wal_level_default = self.common.show_param(self.config_wal_level)
        self.config_slot_log = 'enable_slot_log'
        self.slot_log_default = self.common.show_param(self.config_slot_log)

    def test_logical_replication(self):
        text = "-----step1:修改参数;expect:修改参数成功-----"
        self.logger.info(text)
        if 'on' != self.slot_log_default:
            guc_result = self.sh_primary.execute_gsguc(
                'reload', self.constant.GSGUC_SUCCESS_MSG,
                f'{self.config_slot_log}=on')
            self.assertTrue(guc_result, '执行失败' + text)
        show_msg = self.common.show_param(self.config_slot_log)
        self.assertEquals(show_msg, 'on', '执行失败' + text)

        if 'logical' != self.wal_level_default:
            guc_result = self.sh_primary.execute_gsguc(
                'set', self.constant.GSGUC_SUCCESS_MSG,
                f'{self.config_wal_level}=logical')
            self.assertTrue(guc_result, '执行失败' + text)
            restart_msg = self.sh_primary.restart_db_cluster()
            self.logger.info(restart_msg)
            self.assertTrue(restart_msg, '重启失败')
        show_msg = self.common.show_param(self.config_wal_level)
        self.assertEquals(show_msg, 'logical', '执行失败' + text)

        text = "-----step2:创建表;expect:创建成功-----"
        self.logger.info(text)
        create_table_cmd = f"drop table if exists {self.table_name};" \
              f"create table {self.table_name} (a smallserial, b smallint," \
              f"c int, d bigint, e numeric(5,3) not null, f real not null, " \
              f"g double precision not null,h char(10), i varchar(30), j text," \
              f" k bit varying(20), l timestamp, m date, n boolean not null," \
              f" o json, p tsvector, unique(g,n));"
        create_table_result = self.sh_primary.execut_db_sql(create_table_cmd, dbname='postgres')
        self.logger.info(create_table_result)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS and
                      self.constant.CREATE_TABLE_SUCCESS,
                      create_table_result, '执行失败:' + text)

        text = "-----step3:插入数据;expect:成功-----"
        self.logger.info(text)
        insert_cmd = r'''insert into t_logical_0087
                     (b,c,d,e,f,g,h,i,j,k,l,m,n,o,p) values 
                     (1,2,3,3.54,876.563452345,1.12,'teste',
                     'testando','um texto longo',b'001110010101010',
                     '2013-11-02 17:30:52','2013-02-04','false',
                     '{{\"a\" :123}}','old old parr'::tsvector);
                     insert into t_logical_0087
                     (b,c,d,e,f,g,h,i,j,k,l,m,n,o,p) values 
                     (4,5,6,3.54,876.563452345,4.56,'teste',
                     'testando','um texto longo',b'001110010101010',
                     '2013-11-02 17:30:52','2013-02-04','true',
                     '{{\"a\" :123}}','old old parr'::tsvector);
                     '''
        self.logger.info(insert_cmd)
        insert_result = self.sh_primary.execut_db_sql(insert_cmd, dbname='postgres')
        self.logger.info(insert_result)
        self.assertIn(self.constant.INSERT_SUCCESS_MSG, insert_result, "执行失败" + text)
        query_cmd = f"select 'init' from " \
                    f"pg_create_logical_replication_slot('regression_slot','mppdb_decoding');"
        query_result = self.sh_primary.execut_db_sql(query_cmd, dbname='postgres')
        self.logger.info(query_result)
        self.assertIn('init', query_result, "执行失败" + text)

        text = "-----step4:修改表内索引;expect:成功-----"
        self.logger.info(text)
        update_cmd = f'alter table {self.table_name} replica identity ' \
                     f'using index {self.table_name}_g_n_key;'
        update_result = self.sh_primary.execut_db_sql(update_cmd, dbname='postgres')
        self.logger.info(update_result)
        self.assertIn(self.constant.ALTER_TABLE_MSG, update_result,
            '执行失败:' + text)

        text = "-----step5:修改表内数据;expect:成功-----"
        self.logger.info(text)
        update_cmd = f'update t_logical_0087 set c = -c where b = 1;' \
                     f'update t_logical_0087 set g = -g where n = true;' \
                     f'alter table t_logical_0087 replica identity default;'
        update_result = self.sh_primary.execut_db_sql(update_cmd, dbname='postgres')
        self.logger.info(update_result)
        self.assertIn(self.constant.UPDATE_SUCCESS_MSG, update_result,
                          '执行失败:' + text)

        text = "-----step6:查看数据;expect:成功-----"
        self.logger.info(text)
        sql_cmd1 = f"select data from " \
                   f"pg_logical_slot_peek_changes('regression_slot', null, null);"
        sql_result1 = self.sh_primary.execut_db_sql(sql_cmd1, dbname='postgres')
        self.logger.info(sql_result1)
        self.assertEquals(2, sql_result1.count('"op_type":"UPDATE"'))
        sql_cmd2 = f"select 'stop' from " \
                   f"pg_drop_replication_slot('regression_slot')"
        sql_result2 = self.sh_primary.execut_db_sql(sql_cmd2, dbname='postgres')
        self.logger.info(sql_result2)
        self.assertIn('stop', sql_result2, "执行失败" + text);

    def tearDown(self):
        text = "-----step7:清理表;expect:成功-----"
        self.logger.info(text)
        clear_cmd = f"drop table if exists {self.table_name}"
        clear_result = self.sh_primary.execut_db_sql(clear_cmd, dbname='postgres')
        self.logger.info(clear_result)

        text = "-----step8:恢复参数;expect:成功-----"
        self.logger.info(text)
        slot_log_current = self.common.show_param(self.config_slot_log)
        if slot_log_current != self.slot_log_default:
            self.sh_primary.execute_gsguc(
                'reload', self.constant.GSGUC_SUCCESS_MSG,
                f'{self.config_slot_log}={self.slot_log_default}')
        show_slot_log = self.common.show_param(self.config_slot_log)

        wal_level_current = self.common.show_param(self.config_wal_level)
        if wal_level_current != self.wal_level_default:
            self.sh_primary.execute_gsguc(
                'set', self.constant.GSGUC_SUCCESS_MSG,
                f'{self.config_wal_level}={self.wal_level_default}')
            restart_msg = self.sh_primary.restart_db_cluster()
            self.logger.info(restart_msg)
        show_wal_level = self.common.show_param(self.config_wal_level)
        self.assertIn(show_slot_log, self.slot_log_default,
                          '执行失败' + text)
        self.assertIn(show_wal_level, self.wal_level_default,
                          '执行失败' + text)
        self.logger.info(f'----{os.path.basename(__file__)}:end----')
